package cn.data.stream.stream.inner;


import cn.data.stream.annotation.DataStream;
import cn.data.stream.stream.AbstractStream;
import cn.data.stream.stream.IComponent;
import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.Expression;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

/**
 * 数据分发
 * 配置形如：
 * ${name}.nextStream1=condition1<br>
 * ${name}.nextStream2=condition2
 * @author shi hong yu
 */
@DataStream(value="dispatch")
@Slf4j
public class DispatcherStream extends AbstractStream {

    /**
     *预编译后的分发条件
     * key :nextStream ,value ：编译后的条件
     */
    private Map<String, Expression> dispatchCondition;

    @Override
    public void init() {
        List<IComponent> outputStream = this.getOutputStream();
        if (CollectionUtils.isEmpty(outputStream)) {
            log.warn("{} 无下游，消息将无法分发", this.name);
            return;
        }
        dispatchCondition = new HashMap<>();
        for (IComponent comp : outputStream) {
            String nextName = comp.getName();
            if (this.config.containsProperty(this.name + nextName)) {
                log.warn("{}的下游{}没有配置分发条件，消息将无法分发", this.name, nextName);
            }
            String condition = this.config.getProperty(this.name + nextName);
            if (StringUtils.isBlank(condition)) {
                log.warn("{}的下游{}没有配置分发条件，消息将无法分发", this.name, nextName);
                continue;
            }
            log.info("{} 分发消息到{} 条件:{}", this.name, nextName, condition);
            Expression exp = AviatorEvaluator.compile(condition);
            dispatchCondition.put(nextName, exp);
        }
    }

    @Override
    public void process(Object data) {
        if (data == null) {
            return;
        }
        if (this.getOutputStream() != null) {
            boolean send = false;
            for (IComponent out : this.getOutputStream()) {
                String nextName = out.getName();
                if (!dispatchCondition.containsKey(nextName)) {
                    log.debug("{}的下游{}没有配置分发条件，数据不会分发给它:{}", this.name, nextName, JSON.toJSONString(data));
                    continue;
                }

                Expression expression = dispatchCondition.get(nextName);
                try {
                    Map<String, Object> dataMap = PropertyUtils.describe(data);
                    Boolean match = (Boolean) expression.execute(dataMap);
                    if(match){
                        log.info("数据发往下游：{}->{}", this.name, nextName);
                        out.process(data);
                        send =true;
                    }
                } catch (Exception e) {
                    log.warn("数据发往下游：{}->{} 失败", this.name, nextName, Throwables.getStackTraceAsString(e));
                }
            }
            if(!send){
                log.warn("{} 数据不满足任何分发条件，消息无法分发:{}", this.name,  JSON.toJSONString(data));
            }
        } else {
            log.warn("{} 没有下游组件 数据不下发", this.name);
        }

    }

    @Override
    public void destroy() {
        super.destroy();
    }
}
